package com.kafka.test.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2025/03/20/17:18
 * @Description:
 */

/**
 * 异步发送
 */
public class MyKafkaPartition {
     public static void main(String[] args) {
        //  配置属性集合
        Map<String, Object> configMap = new HashMap<>();
        //  配置属性：Kafka服务器集群地址
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        //  配置属性：Kafka生产的数据为KV对，所以在生产数据进行传输前需要分别对K,V进行对应的序列化操作
        configMap.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        configMap.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        configMap.put(ProducerConfig.ACKS_CONFIG,"-1");  // -1同步 0异步  1：leader节点保存数据之后，就返回发送成功的回调方法
//        配置自定义分区器
        configMap.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyKafkaPartitioner.class.getName());
        //  创建Kafka生产者对象，建立Kafka连接
        //      构造对象时，需要传递配置参数
        KafkaProducer<String, String> producer = new KafkaProducer<>(configMap);
        //  准备数据,定义泛型
        //      构造对象时需要传递 【Topic主题名称】，【Key】，【Value】三个参数
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                    "test", "key" + i, "value" + i
            );
            // TODO 生产（发送）数据  异步发送
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println("数据发送成功");
                }
            });
            System.out.println("消息发生发送成功");
        }
        //  关闭生产者连接
        producer.close();
    }

}
